package com.sihan.framework.etm.common.util;

import com.framework.util.JsonUtils;
import com.sihan.framework.etm.service.baseData.DeviceService;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringEscapeUtils;
import org.json.simple.JSONObject;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory.Will;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

import com.sihan.framework.etm.service.business.ToolPositionService;
import org.springframework.messaging.MessageHeaders;

import java.io.IOException;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

@Configuration
@IntegrationComponentScan
@Slf4j
public class MqttPosReceiveConfig {

    /**
     * 发布的bean名称
     */
    public static final String CHANNEL_NAME_IN = "mqttInputChannel";

    // 客户端与服务器之间的连接意外中断，服务器将发布客户端的“遗嘱”消息
    private static final byte[] WILL_DATA;

    static {
        WILL_DATA = "offline".getBytes();
    }

    @Autowired
    private ToolPositionService toolPositionService;

    @Autowired
    private DeviceService deviceService;

    @Value("${mqtt.username}")
    private String username;

    @Value("${mqtt.password}")
    private String password;

    @Value("${mqtt.pos-host}")
    private String posHostUrl;

    @Value("${mqtt.clientinid}")
    private String clientId;

    @Value("${mqtt.topic}")
    private String defaultTopic;
    @Value("${mqtt.toplevel}")
    private int toplevel;

    @Value("${mqtt.timeout}")
    private int completionTimeout;   //连接超时

    @Bean(name = "posClientFactory")
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        factory.setCleanSession(true);
        factory.setConnectionTimeout(10);
        factory.setKeepAliveInterval(90);
        factory.setPassword(password);
        factory.setServerURIs(posHostUrl);
        factory.setUserName(username);
        factory.setWill(new Will("willTopic", WILL_DATA, 2, false));
        return factory;
    }

    //接收通道
    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }

    //配置client,监听的topic
    @Bean(name = "posInBound")
    public MessageProducer inbound() {
        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter(clientId + "_inbound", mqttClientFactory(),
                        defaultTopic.split(","));
        adapter.setCompletionTimeout(completionTimeout);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(toplevel);
        adapter.setOutputChannel(mqttInputChannel());
        return adapter;
    }

    //通过通道获取数据
    @Bean(name = "posHandler")
    @ServiceActivator(inputChannel = CHANNEL_NAME_IN)
    public MessageHandler handler() {
        return message -> {
            Date date = new Date();
            if (null != message.getHeaders().get("mqtt_topic") &&
                    null != message.getPayload()) {
                Map<String, String> requestMap = new HashMap<>();
                requestMap.put("topic", message.getHeaders().get("mqtt_topic").toString());
                requestMap.put("payload", message.getPayload().toString());
                requestMap.put("time", message.getHeaders().get("timestamp").toString());
                JSONObject jsonObject = new JSONObject(requestMap);
                toolPositionService.addPositionRecord(jsonObject,date);
                deviceService.receiveRecord(jsonObject,date);
            }

        };
    }

}
